package com.its.common.disruptor.factory;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.its.common.disruptor.consumer.BaseMessageConsumer;
import com.its.common.disruptor.dto.TranslatorData;
import com.its.common.disruptor.dto.TranslatorDataWapper;
import com.its.common.disruptor.producer.MessageProducer;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Modifier;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;

/**
 * @Author xiaxp
 * @Date 2020-08-06T18:39:54.729
 * @Description
 */
@Slf4j
public class RingBufferWorkerPoolFactory {

	private static class SingletonHolder {
		static final RingBufferWorkerPoolFactory instance = new RingBufferWorkerPoolFactory();
	}
	
	private RingBufferWorkerPoolFactory(){
		
	}

	private static final String PRODUCER_ID_PREFIX = "disruptor:produceId:";

	private static AtomicInteger MAX_PRODUCER_CNT = new AtomicInteger(1024);
	private static LongAdder currentProducerIndex = new LongAdder();

	public static RingBufferWorkerPoolFactory getInstance() {
		return SingletonHolder.instance;
	}

	public static RingBufferWorkerPoolFactory getInstance(int max) {
		if(max > MAX_PRODUCER_CNT.get()) {
			MAX_PRODUCER_CNT.set(max);
		}
		return SingletonHolder.instance;
	}
	
	private static Map<String, MessageProducer> producers = new ConcurrentHashMap<String, MessageProducer>();
	
	private static Map<String, BaseMessageConsumer> consumers = new ConcurrentHashMap<String, BaseMessageConsumer>();

	private RingBuffer<TranslatorDataWapper> ringBuffer;
	
	private SequenceBarrier sequenceBarrier;
	
	private WorkerPool<TranslatorDataWapper> workerPool;
	
	private void initAndStart(ProducerType type, int bufferSize, WaitStrategy waitStrategy, BaseMessageConsumer[] messageConsumers) {
		//1. 构建ringBuffer对象
		this.ringBuffer = RingBuffer.create(type,
				() -> new TranslatorDataWapper(),
				bufferSize,
				waitStrategy);

		//2.设置序号栅栏
		this.sequenceBarrier = this.ringBuffer.newBarrier();

		//3.设置工作池
		this.workerPool = new WorkerPool<>(this.ringBuffer,
				this.sequenceBarrier,
				new EventExceptionHandler(), messageConsumers);

		//4 把所构建的消费者置入池中
		for(BaseMessageConsumer mc : messageConsumers){
			consumers.put(mc.getConsumerId(), mc);
		}

		//5 添加我们的sequences
		this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());

		//6 启动工作池
		ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
				.setNameFormat("Disruptor-pool-%d").build();

		int coreSize = Runtime.getRuntime().availableProcessors()/2+2;
		if(consumers.size() > coreSize){
			coreSize = consumers.size();
		}

		ExecutorService executorService = new ThreadPoolExecutor(
				coreSize,
				256,
				10,
				TimeUnit.SECONDS,
				new LinkedBlockingQueue<>(1024*1024),
				namedThreadFactory,
				new ThreadPoolExecutor.AbortPolicy());

		this.workerPool.start(executorService);

		addShutDownGracefully(executorService);
	}

	public void initAndStart(BaseMessageConsumer[] consumers) throws Exception {
		if(null == consumers || consumers.length == 0){
			throw new Exception("消费者必须大于0");
		}
		Class<?> clazz = Class.forName(consumers[0].getClass().getName());
		boolean isAbs = Modifier.isAbstract(clazz.getModifiers()) ;
		if(isAbs){
			throw new Exception("消费者类必须是继承于BaseMessageConsumer的非抽象类");
		}
		RingBufferWorkerPoolFactory.getInstance().initAndStart(ProducerType.MULTI,
				1024 * 1024,
				new BlockingWaitStrategy(),
				consumers);
	}
	
	private MessageProducer getMessageProducer(){
		String producerId = PRODUCER_ID_PREFIX + currentProducerIndex.sum();

		MessageProducer messageProducer = producers.get(producerId);
		if(null == messageProducer) {
			messageProducer = new MessageProducer(producerId, this.ringBuffer);
			producers.put(producerId, messageProducer);
		}
		currentProducerIndex.increment();
		if(currentProducerIndex.sum() > MAX_PRODUCER_CNT.get()){
			currentProducerIndex.reset();
		}
		return messageProducer;
	}

	public void produceData(TranslatorData data) {
		if(null == data){
			return;
		}
		MessageProducer messageProducer = getMessageProducer();
		messageProducer.onData(data);
	}


	/**
	 * 异常静态类
	 */
	static class EventExceptionHandler implements ExceptionHandler<TranslatorDataWapper> {
		@Override
		public void handleEventException(Throwable ex, long sequence, TranslatorDataWapper event) {
			log.error("disruptor error, sequence:"+sequence+", event:"+event.getData().getMessage());
		}

		@Override
		public void handleOnStartException(Throwable ex) {
			log.error("handleOnStartException: "+ex);
		}

		@Override
		public void handleOnShutdownException(Throwable ex) {
			log.error("handleOnShutdownException: "+ex);
		}
	}

	/**
	 * 优雅关闭
	 */
	private void addShutDownGracefully(ExecutorService executorService) {
		Runtime.getRuntime().addShutdownHook(new Thread(() -> {
			try {
				executorService.shutdown();
				Thread.sleep(500L);
			} catch (InterruptedException e) {
			}
		}));
	}
}



